package com.pl.study.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;

/**
 * <p>
 *
 * @Description: hbase test
 * </p>
 * @ClassName HBaseTest
 * @Author pl
 * @Date 2022/3/19
 * @Version V1.0.0
 */
public class HBaseTest {
    private static final String MY_NAME = "penglei";
    private static final String MY_TABLE_NAME = MY_NAME + ":student";
    private static final String MY_STUDENT_ID = "G20220735020168";
    private static final String MY_CLASSS = "1";
    private static final String MY_UNDERSTANDING = "100";
    private static final String MY_PROGRAMMING = "100";
    private static final String FAMILYNAMES[] = {"nature", "social"};
    private static final Logger logger = LoggerFactory.getLogger(HBaseTest.class);

    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;

    public static void main(String[] args) throws IOException {
        init();
        TableName tableName = TableName.valueOf(MY_TABLE_NAME);

        Table table = createTable(tableName);

        addData(table);

        getData(table);

        deleteData(table);

        getData(table);

        dropTable(tableName);

        logger.info("process ended...");
    }

    public static void init() {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", " emr-worker-2.cluster-285604,emr-worker-1.cluster-285604,emr-header-1.cluster-285604");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    private static Table createTable(TableName tableName) throws IOException {
        NamespaceDescriptor[] listNamespaceDescriptors  = admin.listNamespaceDescriptors();
        logger.info("all namespace :{}", Arrays.toString(listNamespaceDescriptors));
        long count = Arrays.stream(listNamespaceDescriptors).filter(data -> data.equals(MY_NAME)).count();
        if (count==0){
            admin.createNamespace(NamespaceDescriptor.create(MY_NAME).build());
            logger.info("namespace: {},created",MY_NAME);
        }

        if (!admin.tableExists(tableName)) {
            createTable(tableName,FAMILYNAMES);
        }

        return connection.getTable(tableName);
    }


    public static void createTable(TableName tableName, String familyNames[]) throws IOException {
        if (admin.tableExists(tableName)) {
            System.out.println("table exist");
            return;
        }
        TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
        for (String familyName : familyNames) {
            ColumnFamilyDescriptor of = ColumnFamilyDescriptorBuilder.of(familyName);
            builder.setColumnFamily(of);
        }
        TableDescriptor build = builder.build();
        admin.createTable(build);
        logger.info("table created...");
    }



    private static void addData(Table table) throws IOException {
        Put put = new Put(Bytes.toBytes(MY_NAME));

        put.addColumn(Bytes.toBytes(FAMILYNAMES[0]), Bytes.toBytes("student_id"), Bytes.toBytes(MY_STUDENT_ID));
        put.addColumn(Bytes.toBytes(FAMILYNAMES[0]), Bytes.toBytes("class"), Bytes.toBytes(MY_CLASSS));

        put.addColumn(Bytes.toBytes(FAMILYNAMES[1]), Bytes.toBytes("understanding"), Bytes.toBytes(MY_UNDERSTANDING));
        put.addColumn(Bytes.toBytes(FAMILYNAMES[1]), Bytes.toBytes("programming"), Bytes.toBytes(MY_PROGRAMMING));

        table.put(put);
        logger.info("data added...");
    }

    private static void getData(Table table) throws IOException {
        Get get = new Get(Bytes.toBytes(MY_NAME));
        if (!get.isCheckExistenceOnly()) {
            Result result = table.get(get);
            for (Cell cell : result.rawCells()) {
                String col = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String val = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                logger.info("Read data ==> col:{}, val: {}",col,val);
            }
        }
    }

    private static void deleteData(Table table) throws IOException {
        Delete delete = new Delete(Bytes.toBytes(MY_NAME));
        table.delete(delete);
    }

    private static void dropTable(TableName tableName) throws IOException {
        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            logger.info("table deleted...");
        } else {
            logger.info("table not exists...");
        }
    }
}